package recommendation.book.ml

import javax.xml.transform.stream.StreamResult

import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.recommendation.ALS
import org.apache.spark.ml.recommendation.ALS.Rating
import org.apache.spark.sql.SparkSession
import org.dmg.pmml.FieldName
import org.jpmml.evaluator.ModelEvaluatorFactory
import org.jpmml.model.JAXBUtil
import org.jpmml.sparkml.ConverterUtil

import scala.collection.JavaConversions._

/**
  * Created by peibin on 2017/4/13.
  */
object ALSPipeLineDemo {

  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .appName("WeiboFeeds")
      .master("local[*]")
      .getOrCreate()

    import spark.implicits._


    val ratings = spark.read.option("header", "true").option("delimiter", ";").csv("""/Users/peibin/workspace/data-mining/书籍评分/BX-Book-Ratings.csv""")
      .map(x => Rating(x.getString(0).toInt, x.getString(1).hashCode, x.getString(2).toFloat))
    val als = new ALS()
      .setRank(5)
      .setMaxIter(10)
    val pipeline = new Pipeline().setStages(Array(als));
    val model = pipeline.fit(ratings)
    val pmml = ConverterUtil.toPMML(ratings.schema, model);
    JAXBUtil.marshalPMML(pmml, new StreamResult(System.out));
    val request: Map[FieldName, _] = Map()
    ModelEvaluatorFactory.newInstance.newModelEvaluator(pmml).evaluate(request)

    spark.stop()

  }


}
